package com.navinfo.platform.bigdata.hive.udtf;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.navinfo.platform.bigdata.hive.dto.VehicleStatusAdditionDto;
import static com.navinfo.platform.bigdata.hive.dto.LCStatusType.StatusType.*;
import org.apache.hadoop.hive.ql.exec.Description;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

import java.util.ArrayList;

/**
 * 解析状态附加位的信息，返回相应字段
 * 一对多类型：接受零个或多个输入，然后产生多列或多行的输出
 * UDTF：用户自定义表生成函数，如explode()，需要继承org.apache.hadoop.hive.ql.udf.generic.GenericUDTF；
 */
@Description(
	name = "status_addition",
	value = "_FUNC_(col) - The parameter is statusaddition column. The return value is 16 strings.",
	extended = "Example:\n"
		+ " > SELECT _FUNC_(col) FROM src;"
		+ " > SELECT _FUNC_(col) AS (torque, rotation,accelerator) FROM src;"
		+ " > SELECT adTable.torque,adTable.rotation FROM src LATERAL VIEW _FUNC_(col) adTable AS name, surname;"
)
public class StatusAdditionUDF extends GenericUDTF{
    private String[] result;
	/**
     * 该方法指定输入输出参数：输入的Object Inspectors和输出的Struct。
     * @param argOIs
     * @return
     * @throws UDFArgumentException
     */
	@Override
	public StructObjectInspector initialize(ObjectInspector[] argOIs) throws UDFArgumentException {
		if(argOIs.length != 1){
			throw new UDFArgumentException("StatusAdditionUDF takes exactly one argument.");
		}

		if(argOIs[0].getCategory() != ObjectInspector.Category.PRIMITIVE
				&& ((PrimitiveObjectInspector)argOIs[0]).getPrimitiveCategory() != PrimitiveObjectInspector.PrimitiveCategory.STRING){
			throw new UDFArgumentTypeException(0, "StatusAdditionUDF takes a string as a parameter.");
		}
		
		ArrayList<String> fieldNames = new ArrayList<>();
        fieldNames.add("torque");
		fieldNames.add("rotation");
		fieldNames.add("accelerator");
		fieldNames.add("clutchSwitch");
		fieldNames.add("realtimeFuelConsumption");
		fieldNames.add("fuelConsumptionRate");
		fieldNames.add("ecuMileage");
		fieldNames.add("meterMileage");
		fieldNames.add("integralMileage");
		fieldNames.add("integralFuelCons");
		fieldNames.add("totalFuelCons");
		fieldNames.add("coolingWaterTem");
		fieldNames.add("intakeAirTem");
		fieldNames.add("intakePressure");
		fieldNames.add("cruiseControl");
		fieldNames.add("oilPercentage");
		fieldNames.add("brakeSwitch");
		fieldNames.add("oilPressure");
		fieldNames.add("atmosphericTem");

		ArrayList<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>();
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);
	}

    /**
     * 该方法处理输入记录，然后通过forward()方法返回输出结果
     * @param args
     * @throws HiveException
     */
	@Override
	public void process(Object[] args) throws HiveException {
		result = new String[]{
				"0","0","0","0","0",
				"0","0","0","0","0",
				"0","0","0","0","0",
				"0","0","0","0"};
		System.out.println(args[0].getClass());
		String input = args[0].toString();

		//原始值都扩大了100倍，没处理
		for (VehicleStatusAdditionDto dto : VehicleStatusAdditionDto.parseToList(input)) {
			String statusValue = dto.getStatusValue();
			switch(Integer.parseInt(dto.getTypes())) {
				case actualEnginePercentTor_VALUE:
					//发动机输出扭矩，单位%，
					result[0] = statusValue;
					break;
				case rotation_VALUE:
					//发动机转速，单位rpm转每分钟
					result[1] = statusValue;
					break;
				case accPedalPos_VALUE:
					//油门开度百分比，单位%
					result[2] = statusValue;
					break;
				case clutchSwitch_VALUE:
					//离合开关状态，0关，1开
					result[3] = statusValue;
					break;
				case realTimeOilConsumption_VALUE:
					//实时油耗(发动机瞬时油耗)（KM/L）
					result[4] = statusValue;
					break;
				case fuelConsumptionRate_VALUE:
					//发动机燃油消耗率（毫升/小时）转为 L/H
					result[5] = String.valueOf(Long.valueOf(statusValue)/1000.0);
					break;
				case mileage_VALUE:
					//can里程(ecu)，单位KM
					result[6] = statusValue;
					break;
				case mileageDD_VALUE:
					//仪表里程，单位KM
					result[7] = statusValue;
					break;
				case differentialMileage_VALUE:
					//积分里程，单位KM
					result[8] = statusValue;
					break;
				case integralFuelConsumption_VALUE:
					//积分油耗，单位L
					result[9] = statusValue;
					break;
				case totalFuelConsumption_VALUE:
					//燃油总消耗，单位L
					result[10] = statusValue;
					break;
				case coolingWaterTem_VALUE:
					//冷却水温,发动机冷却液问题
					result[11] = statusValue;
					break;
				case intakeAirTem_VALUE:
					//进气温度
					result[12] = statusValue;
					break;
				case dpfPressure_VALUE:
					//进气压力
					result[13] = statusValue;
					break;
				case cruiseCtrlActive_VALUE:
					//定速巡航开关
					result[14] = statusValue;
					break;
				case oilValue_VALUE:
					//油液百分比,燃油液位
					result[15] = statusValue;
					break;
				case brakeSwitch_VALUE:
					result[16] = statusValue;
				case oilPressure_VALUE:
					//机油压力
					result[17] = statusValue;
				case atmosphericTem_VALUE:
					//环境温度
					result[18] = statusValue;
				default:
					break;
			}
		}

		forward(result);
	}

    /**
     *
     * 用于通知UDTF没有行可以处理了。可以在该方法中清理代码或者附加其他处理输出。
     * @throws HiveException
     */
	@Override
	public void close() throws HiveException {
		result = null;
	}

}
